package hadoop.ch06.v17034460231;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.List;
import java.util.Map;

public class HBaseDemo63 {
    public static void main(String[] args)throws Exception{
        createTable();
        insert();
        get();
        scan();
    }
    /**l
     * 创建表格
     * @throws Exception
     */
   private static void createTable() throws Exception{
       //指定ZooKeeper地址，从ZK中获取HMaster的地址
       Configuration conf = HBaseConfiguration.create();
       //完全分布式的HBase/conf/hbase-site.xml的配置
       conf.set("hbase.zookeeper.quorum","node1,node2,node3");
       conf.set("hbase.rootdir","hdfs://node1:8020/hbase");
       conf.set("hbase.cluster.distributed","true");
       conf.set("hbase.zookeeper.property.clientPort","2181");
       //创建到HBase的链接
       Connection connect = ConnectionFactory.createConnection(conf);
       Admin admin = connect.getAdmin();
       //System.out.println(connect);
       //System.out.println(admin);
       //Table的名称
       TableName tn = TableName.valueOf("emp17034460231");
       //Table的列族
       String[] family = new String[]{"member_id","address","info"};
       //Table的描述，把列族加入描述当中
       HTableDescriptor desc = new HTableDescriptor(tn);
       for (int i = 0; i < family.length; i++) {
           desc.addFamily(new HColumnDescriptor(family[i]));
       }
       if (admin.tableExists(tn)) {
           //已存在则提示
           System.out.println("table Exists!");
          // System.exit(0);
       } else {
           //不存在则创建
           admin.createTable(desc);
           System.out.println("create table Success!");
           }
       //System.out.println("执行结束!");
       }
    /**
     * 实现此方法，往表里插入数据
     * @throws Exception
     */
    private static void insert() throws Exception {
        //指定ZooKeeper地址，从ZK中获取HMaster的地址
        Configuration conf = HBaseConfiguration.create();
        //完全分布式的HBase/conf/hbase-site.xml的配置
        conf.set("hbase.zookeeper.quorum","node1,node2,node3");
        conf.set("hbase.rootdir","hdfs://node1:8020/hbase");
        conf.set("hbase.cluster.distributed","true");
        //创建到HBase的链接
        Connection connect = ConnectionFactory.createConnection(conf);
        /* get table. 指定表的客户端*/
        TableName tn = TableName.valueOf("emp17034460231");
        Table table = connect.getTable(tn);

        /* create put. */
        //行键
        String rowKey = new String("Rain");
        Put put = new Put(Bytes.toBytes(rowKey));
        //String[] column = new String[]{"member_id","info:age","info:birthday","info:industry","info:city","info:country"};
        //String[] value = new String[]{"31","28","1990-05-01","architect","ShenZhen","China"};
        put.addColumn(Bytes.toBytes("member_id"),    //列族的名字
                Bytes.toBytes(" "),           //列的名字
                Bytes.toBytes("31"));         //值

        put.addColumn(Bytes.toBytes("info"),    //列族的名字
                Bytes.toBytes("age"),           //列的名字
                Bytes.toBytes("28"));         //值

        put.addColumn(Bytes.toBytes("info"),    //列族的名字
                Bytes.toBytes("birthday"),           //列的名字
                Bytes.toBytes("1990-05-01"));         //值

        put.addColumn(Bytes.toBytes("info"),    //列族的名字
                Bytes.toBytes("industry"),           //列的名字
                Bytes.toBytes("architect"));         //值

        put.addColumn(Bytes.toBytes("info"),    //列族的名字
                Bytes.toBytes("city"),           //列的名字
                Bytes.toBytes("ShenZhen"));         //值

        put.addColumn(Bytes.toBytes("info"),    //列族的名字
                Bytes.toBytes("country"),           //列的名字
                Bytes.toBytes("China"));         //值

        /* put data. */
        table.put(put);
        System.out.println("------------------------------------");
        System.out.println("add data Success!");
    }
    /**
      * 实现此方法，查询数据，
      * 并打印查询结果到控制台（System,.out.println)
      * @throws Exception
     **/
    private static void get() throws Exception {
        //指定ZooKeeper地址，从ZK中获取HMaster的地址
        Configuration conf = HBaseConfiguration.create();
        //完全分布式的HBase/conf/hbase-site.xml的配置
        conf.set("hbase.zookeeper.quorum","node1,node2,node3");
        conf.set("hbase.rootdir","hdfs://node1:8020/hbase");
        conf.set("hbase.cluster.distributed","true");
        //创建到HBase的链接
        Connection connect = ConnectionFactory.createConnection(conf);
        /* get table.指定表的客户端 */
        Table table = connect.getTable(TableName.valueOf("emp17034460231"));
        //通过get查询
        String rowKey = new String("Rain");
        Get get = new Get(Bytes.toBytes(rowKey));
        // 获取指定列族和列修饰符对应的列
        //get.addColumn(Bytes.toBytes("info"),Bytes.toBytes(""));
        //执行查询
        Filter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator("info".getBytes()));
        get.setFilter(familyFilter);
        Result result = table.get(get);
        //获取列族info的所有列
/*        Map<byte[], byte[]> familyMap = result.getFamilyMap(Bytes.toBytes("info"));
        for(Map.Entry<byte[], byte[]> entry:familyMap.entrySet()){
            System.out.println(Bytes.toString(entry.getKey()));
            }*/
        //输出
        //String name = Bytes.toString(result.getValue(Bytes.toBytes("info"),null));
        //System.out.println(name);
        for (Cell cell : result.listCells()) {
                System.out.println("------------------------------------");
                System.out.println("rowkey: " + new String(CellUtil.cloneRow(cell)));
                System.out.println("family: " + new String(CellUtil.cloneFamily(cell)));
                System.out.println("column: " + new String(CellUtil.cloneQualifier(cell)));
                System.out.println("value : " + new String(CellUtil.cloneValue(cell)));
                System.out.println("timest: " + cell.getTimestamp());
            }
        System.out.println("get over!");
    }
    /**
     * 实现此方法，按实验要求扫描表数据，
     * 并打印查询结果到控制台（System,.out.println)
     * @throws Exception
     */
    private static void scan() throws Exception{
        //指定ZooKeeper地址，从ZK中获取HMaster的地址
        Configuration conf = HBaseConfiguration.create();
        //完全分布式的HBase/conf/hbase-site.xml的配置
        conf.set("hbase.zookeeper.quorum","node1,node2,node3");
        conf.set("hbase.rootdir","hdfs://node1:8020/hbase");
        conf.set("hbase.cluster.distributed","true");
        //创建到HBase的链接
        Connection connect = ConnectionFactory.createConnection(conf);
        /* get table.指定表的客户端 */
        Table table = connect.getTable(TableName.valueOf("emp17034460231"));
        //创建一个扫描器Scan
        Scan scan = new Scan();   //相当于：select * from emp17034460231
        //以下代码查询列的名字为birthday，但是列的名字为无（空格除外）的也同样查询出来
        //列过滤器 QualifierFilter
        Filter qualifierFilter = new QualifierFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator("birthday".getBytes()));
        scan.setFilter(qualifierFilter);
        ResultScanner resultScanner = table.getScanner(scan);
        for (Result result:resultScanner){
           //List<Cell>cells = result.listCells();
           // for (Cell cell:cells){
            for (Cell cell:result.listCells()){
                //System.out.println(cell);
                System.out.println("------------------------------------");
                System.out.println("rowkey: " + new String(CellUtil.cloneRow(cell)));
                System.out.println("family: " + new String(CellUtil.cloneFamily(cell)));
                System.out.println("column: " + new String(CellUtil.cloneQualifier(cell)));
                System.out.println("value : " + new String(CellUtil.cloneValue(cell)));
                System.out.println("timest: " + cell.getTimestamp());
            }
        }
        System.out.println("scan over!");
    }
}
